package com.tencent.cloud.iov.util.rx;

import io.reactivex.Notification;
import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.functions.Function;

/**
 * 消息转换工具类
 *
 * 结合错误进行装箱 / 拆箱进行使用，可避免类似 Subject 遇到错误后，订阅关系不再生效的场景
 *
 * @author showxiao
 * @date 2018/4/24
 */

public class ErrorNextTransformer {

    /**
     * 将数据或者错误包装成 Notification
     *
     * @param <T>       原数据类型
     * @return
     */
    public static <T> ObservableTransformer<T, Notification<T>> onErrorNext() {
        return new OnErrorNextTransformer<>();
    }

    /**
     * 从 Notification 提取数据或者错误
     *
     * @param <T>       原数据类型
     * @return
     */
    public static <T> ObservableTransformer<Notification<T>, T> onNextError() {
        return new OnNextErrorTransformer<>();
    }

    public static <T> ObservableTransformer<T, T> onErrorEmpty() {
        return new OnErrorEmptyTransformer<>();
    }

    /**
     * 将数据或者错误包装成 Notification
     *
     * @param <T>       原数据类型
     */
    private static class OnErrorNextTransformer<T>
            implements ObservableTransformer<T, Notification<T>> {

        @Override
        public ObservableSource<Notification<T>> apply(Observable<T> upstream) {
            return upstream
                    .map(new Function<T, Notification<T>>() {
                        @Override
                        public Notification<T> apply(T t) throws Exception {
                            return Notification.createOnNext(t);
                        }
                    })
                    .onErrorResumeNext(new Function<Throwable, ObservableSource<? extends Notification<T>>>() {
                        @Override
                        public ObservableSource<? extends Notification<T>> apply(Throwable throwable) throws Exception {
                            return Observable.just(Notification.<T>createOnError(throwable));
                        }
                    });
        }
    }

    /**
     * 从 Notification 提取数据或者错误
     *
     * @param <T>       原数据类型
     */
    private static class OnNextErrorTransformer<T>
            implements ObservableTransformer<Notification<T>, T> {

        @Override
        public ObservableSource<T> apply(Observable<Notification<T>> upstream) {
            return upstream
                    .flatMap(new Function<Notification<T>, ObservableSource<T>>() {
                        @Override
                        public ObservableSource<T> apply(Notification<T> notification) throws
                                Exception {
                            if (notification.isOnError()) {
                                return Observable.error(notification.getError());
                            }

                            return Observable.just(notification.getValue());
                        }
                    });
        }
    }

    private static class OnErrorEmptyTransformer<T> implements ObservableTransformer<T, T> {

        @Override
        public ObservableSource<T> apply(Observable<T> upstream) {
            return upstream
                    .onErrorResumeNext(new Function<Throwable, ObservableSource<? extends T>>() {
                        @Override
                        public ObservableSource<? extends T> apply(Throwable throwable) throws Exception {
                            return Observable.empty();
                        }
                    });
        }
    }
}
